package 生产者;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import 自定义分区器.Kafka01_MyPartition;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * Kafka生产者可能会遇到数据重复和乱序的问题，这些问题通常与Kafka的工作机制、配置以及网络状况等因素有关。下面分别解释这两种情况的原因及可能的解决方案。
 *
 * 数据重复
 * Producer Retry：当生产者发送消息失败时（例如由于网络问题），它可能会根据配置重试发送该消息。如果此时broker已经接收了消息但未及时返回确认给生产者（可能是由于网络延迟或broker暂时不可用），那么重试操作可能导致消息被多次写入Kafka。
 * 解决方案：可以使用幂等性生产者来避免这个问题。通过设置enable.idempotence=true，Kafka可以在一定程度上保证消息的唯一性，即使在网络异常情况下也不会导致重复消息。
 * Consumer Processing Failure：消费者处理消息失败后重新消费相同的消息也会造成数据重复。比如，消费者在处理完消息之后提交offset之前崩溃了，那么重启后会从上次已知的offset开始重新消费消息。
 * 解决方案：确保消费者的offset提交逻辑是可靠的，并考虑实现基于业务逻辑的去重机制。
 * 乱序
 * 异步发送和批次处理：Kafka生产者默认采用异步方式发送消息，并且会将多个小的消息合并成一个批次(batch)进行发送。这可能导致不同批次之间的消息顺序被打乱，特别是当设置了linger.ms参数以增加批处理效率时。
 * 解决方案：对于需要严格顺序的消息，可以通过设置max.in.flight.requests.per.connection=1来强制单个连接上的请求串行化执行，但这会影响吞吐量。另一种方法是使用消息键(message key)，让具有相同键的消息总是被发送到同一个分区(partition)，从而保持它们的相对顺序。
 * 分区策略：如果消息没有指定具体的分区而是依赖于默认的分区器(partitioner)，那么不同的消息可能会因为负载均衡等原因被分配到不同的分区，这也可能导致最终读取时的顺序与原始发送顺序不一致。
 * 解决方案：如前所述，为需要保持顺序的消息指定相同的key，或者直接指定目标分区，这样可以确保这些消息不会被分散到不同的分区中。
 * Leader副本切换：在ISR机制下，如果leader副本发生故障并触发选举新的leader，可能会出现短暂的时间窗口，在此期间follower副本还未完全同步所有消息，导致部分消息顺序混乱。
 * 解决方案：虽然这种情况难以完全避免，但是可以通过调整ISR相关的参数来减少影响，例如降低replica.lag.time.max.ms值，使follower更紧密地跟随leader。
 * 理解上述原因有助于合理设计Kafka的应用场景，选择合适的配置选项，以满足对数据重复性和顺序性的要求。
 *
 *
 */
public class Kafka02_Productor2 {
    public static void main(String[] args) {

// 配置Kafka生产者的属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "114.251.235.19:9092"); // Kafka服务器地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, Kafka01_MyPartition.class);//此处配置自定义分区其

        // 创建Kafka生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 要发送的消息和目标主题
        String topic = "topic_test";
        String key = "messageKey";
        String value = "Hello, Kafka!";

        // 创建ProducerRecord对象
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        try {
            // 发送消息并等待结果（同步发送）
            RecordMetadata metadata = producer.send(record).get();

            // 打印消息发送的结果
            System.out.println("Message sent to topic:" + metadata.topic() +
                    ", partition:" + metadata.partition() +
                    ", offset:" + metadata.offset());
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}
